#include "pg_school.h"

#include "postgres.h"
#include "fmgr.h"
#include "funcapi.h"
#include "executor/spi.h"
#include "utils/builtins.h"
#include "utils/palloc.h"
#include "utils/rel.h"
#include "catalog/pg_class.h"
#include "catalog/pg_namespace.h"
#include "access/heapam.h"
#include "access/table.h"
#include "utils/array.h"
#include "utils/geo_decls.h"
#include "utils/typcache.h"
#include "access/genam.h"

PG_MODULE_MAGIC;

void my_exec_spi(OperationType opType, char *sql)
{
    int ret = 0;

    elog(INFO, "EXECUTE SQL: %s", sql);
    ret = SPI_exec(sql, 0);

    switch (opType) {
        case OP_INSERT:
            if (ret!= SPI_OK_INSERT)
                elog(ERROR, "Insert operation failed: %s", sql);
            break;
        case OP_DELETE:
            if (ret!= SPI_OK_DELETE)
                elog(ERROR, "Delete operation failed: %s", sql);
            break;
        case OP_UPDATE:
            if (ret!= SPI_OK_UPDATE)
                elog(ERROR, "Update operation failed: %s", sql);
            break;
        case OP_QUERY:
            if (ret!= SPI_OK_SELECT)
                elog(ERROR, "Query operation failed: %s", sql);
            break;
    }

}

PG_FUNCTION_INFO_V1(insert_score_spi);

Datum insert_score_spi(PG_FUNCTION_ARGS)
{
    ArrayType *input_array = PG_GETARG_ARRAYTYPE_P(0);
    Datum *elements = NULL;
    bool *nulls = NULL;
    int num_elements = 0;
    int ret = 0;
    uint64 total_rows = 0;
    is_udf_execution = true;
    // 检查 input_array 是否为 NULL
    if (input_array == NULL || ARR_NDIM(input_array) == 0)
    {
        is_udf_execution = false;
        elog(WARNING, "Input array is NULL or empty");
        PG_RETURN_INT32(0);
    }

    deconstruct_array(input_array,
                      TEXTOID,
                      -1, false, 'i',
                      &elements, &nulls, &num_elements);
    // 检查元素数量
    if (num_elements == 0)
    {
        is_udf_execution = false;
        elog(WARNING, "Input array has no elements");
        PG_RETURN_INT32(0);
    }

    ret = SPI_connect();
    if (ret != SPI_OK_CONNECT)
    {
        is_udf_execution = false;
        elog(WARNING, "SPI CONNECT ERROR");
        PG_RETURN_INT32(0);
    }

    for (int i = 0; i < num_elements; i++)
    {
        char *tuple_str = NULL;
        char *temp = NULL;
        char *query = NULL;
        char *check_unique_query = NULL;
        char *check_student_query = NULL;
        char id_str[10], ranking_str[10];
        int parsed = 0;
        if (nulls[i])
        {
            elog(ERROR, "ARGS EXIST NULL VALUE");
        }

        tuple_str = TextDatumGetCString(elements[i]);
        //解析字符串，并忽略空格
        parsed  = sscanf(tuple_str, "( %*[^,], %*[^,], \"%*[^\"]\", %[^,], %[^)])", id_str, ranking_str);
        //如果解析出的参数不是两个，告警并跳过这次插入
        if (parsed != 2)
        {
            elog(WARNING, "Invalid input format for tuple: %s", tuple_str);
            pfree(tuple_str);
            continue;
        }

        // 检查另一个相关表中是否存在相同的 id
        check_student_query = psprintf("SELECT 1 FROM %s WHERE id = %s", SCHEMATABLESTUDENT, id_str);
        ret = SPI_execute(check_student_query, false, 0);
        if (SPI_processed == 0)
        {
            // 如果 student 表中不存在该 id，发出警告并跳过插入
            elog(WARNING, "No student found with id: %s, skipping insertion", id_str);
            pfree(tuple_str);
            pfree(check_student_query);
            continue;
        }
        pfree(check_student_query);

        // 检查 id 或者ranking是否唯一
        check_unique_query = psprintf(" SELECT 1 FROM %s WHERE id = %s OR ranking = '%s'",\
                                     SCHEMATABLESCORE, id_str, ranking_str );
        ret = SPI_execute(check_unique_query, false, 0);
        if (SPI_processed > 0)
        {
            // 如果不唯一，发出警告并跳过这次插入
            elog(WARNING, "Duplicate id or ranking found: %s,%s, skipping insertion", id_str, ranking_str);
            pfree(tuple_str);
            pfree(check_unique_query);
            continue;
        }

        temp = tuple_str;
        while (*temp) {
            if (*temp == '"')
                *temp = '\'';
            temp++;
        }
        query = psprintf("INSERT INTO %s VALUES %s", SCHEMATABLESCORE, tuple_str);
        // 执行查询
        my_exec_spi(OP_INSERT, query);
        CommandCounterIncrement();
        total_rows += SPI_processed;
        pfree(query);
        pfree(tuple_str);
    }

    // 记录总插入行数
    log_operateRows(TABLESCORE, insertType, total_rows);

    // 完成 SPI 连接
    ret = SPI_finish();
    if (ret != SPI_OK_FINISH)
    {
        is_udf_execution = false;
        elog(ERROR, "SPI FINISH ERROR");
    }
    is_udf_execution = false;
    PG_RETURN_INT32(total_rows);
}


PG_FUNCTION_INFO_V1(delete_score_spi);

Datum delete_score_spi(PG_FUNCTION_ARGS)
{
    char *query = NULL;
    char *something_to_delete = NULL;
    int actual_rows = 0;
    is_udf_execution = true;//是否在执行UDF函数
    // 获取并校验传入参数
    if (PG_ARGISNULL(0))
    {
        is_udf_execution = false;
        elog(WARNING, "Input argument is NULL");
        PG_RETURN_INT32(0);
    }
    something_to_delete = text_to_cstring(PG_GETARG_TEXT_PP(0));
    // 检查传入的条件是否为空字符串
    if (strlen(something_to_delete) == 0)
    {
        is_udf_execution = false;
        elog(WARNING, "The delete condition cannot be an empty string");
        PG_RETURN_INT32(0);
    }
    // 检查条件中是否包含有效的比较运算符（>、<、=）
    if (strchr(something_to_delete, '>') == NULL &&
        strchr(something_to_delete, '<') == NULL &&
        strchr(something_to_delete, '=') == NULL)
    {
        is_udf_execution = false;
        elog(WARNING, "Invalid format for delete condition, must include one of '>', '<', or '='");
        PG_RETURN_INT32(0);
    }

    query = psprintf("DELETE FROM %s WHERE %s", SCHEMATABLESCORE, something_to_delete);
    elog(INFO, "delete query: %s", query);
    SPI_connect();
    my_exec_spi(OP_DELETE, query);
    actual_rows = SPI_processed;
    log_operateRows(TABLESCORE,deleteType,actual_rows);
    SPI_finish();
    pfree(query);
    is_udf_execution = false;
    PG_RETURN_INT32(actual_rows);
}

PG_FUNCTION_INFO_V1(update_score_spi);

Datum update_score_spi(PG_FUNCTION_ARGS)
{
    char *query = NULL;
    char *col_name = text_to_cstring( PG_GETARG_TEXT_PP(0));
    char *cmp_operator = text_to_cstring(PG_GETARG_TEXT_PP(1));
    char *col_value = text_to_cstring( PG_GETARG_TEXT_PP(2));
    char *update_col_name = text_to_cstring( PG_GETARG_TEXT_PP(3));
    char *update_col_value = text_to_cstring( PG_GETARG_TEXT_PP(4));
    int actual_rows = 0;
    char *check_unique_query = NULL;
    int ret = 0;
    is_udf_execution = true;//是否在执行UDF函数
    // 校验 cmp_operator 是否为有效的比较运算符
    if (strcmp(cmp_operator, ">") != 0 && strcmp(cmp_operator, "<") != 0 && strcmp(cmp_operator, "=") != 0)
    {
        elog(WARNING, "Invalid comparison operator. Only '>', '<', '=' are allowed.");
        is_udf_execution = false;
        PG_RETURN_INT32(0);
    }

    // 校验其他输入是否为空
    if (strlen(col_name) == 0 || strlen(col_value) == 0 || strlen(update_col_name) == 0 || strlen(update_col_value) == 0)
    {
        elog(WARNING, "Column names and values cannot be empty.");
        is_udf_execution = false;
        PG_RETURN_INT32(0);
    }

    // 如果更新的是 id 或 ranking，检查是否唯一
    if (strcmp(update_col_name, "id") == 0 || strcmp(update_col_name, "ranking") == 0)
    {
        // 生成检查唯一性的查询
        check_unique_query = psprintf("SELECT 1 FROM %s WHERE %s = '%s'", SCHEMATABLESCORE, update_col_name, update_col_value);

        ret = SPI_connect();
        if (ret != SPI_OK_CONNECT)
        {
            elog(WARNING, "SPI CONNECT ERROR");
            is_udf_execution = false;
            PG_RETURN_INT32(0);
        }

        // 执行唯一性检查查询
        ret = SPI_execute(check_unique_query, true, 0);
        if (ret != SPI_OK_SELECT)
        {
            elog(WARNING, "Error executing uniqueness check query");
            SPI_finish();
            is_udf_execution = false;
            PG_RETURN_INT32(0);
        }

        // 如果存在结果，则跳过更新
        if (SPI_processed > 0)
        {
            elog(WARNING, "Duplicate value found for %s: %s. Skipping update.", update_col_name, update_col_value);
            SPI_finish();
            is_udf_execution = false;
            PG_RETURN_INT32(0);
        }

        pfree(check_unique_query);
        SPI_finish();
    }

    query = psprintf("UPDATE %s SET %s = '%s' WHERE %s %s '%s'",\
            SCHEMATABLESCORE, update_col_name, update_col_value, col_name, cmp_operator, col_value);   
    SPI_connect();
    my_exec_spi(OP_UPDATE, query);
    actual_rows = SPI_processed;
    log_operateRows(TABLESCORE,updateType,actual_rows);
    SPI_finish();
    pfree(query);
    is_udf_execution = false;
    PG_RETURN_INT32(actual_rows);
}

PG_FUNCTION_INFO_V1(query_score_spi);

Datum query_score_spi(PG_FUNCTION_ARGS) {
    char *col_name = text_to_cstring(PG_GETARG_TEXT_PP(0));
    char *cmp_operator = text_to_cstring(PG_GETARG_TEXT_PP(1));
    char *input_value = text_to_cstring(PG_GETARG_TEXT_PP(2));
    char *query = NULL;
    int ret = -1;
    ReturnSetInfo *rsinfo = NULL;
    // 校验 cmp_operator 是否为有效的比较运算符
    if (strcmp(cmp_operator, ">") != 0 && strcmp(cmp_operator, "<") != 0 && strcmp(cmp_operator, "=") != 0)
    {
        elog(WARNING, "Invalid comparison operator. Only '>', '<', '=' are allowed.");
        return (Datum) 0;
    }
    // 校验其他输入是否为空
    if (strlen(col_name) == 0 || strlen(input_value) == 0 )
    {
        elog(WARNING, "Column names and values cannot be empty.");
        return (Datum) 0;
    }
#define PG_STAT_GET_SCORE_COLS 6
    rsinfo = (ReturnSetInfo *)fcinfo->resultinfo;
    InitMaterializedSRF(fcinfo, 0);

    // Construct the SQL query
    query = psprintf("SELECT * FROM %s WHERE %s %s '%s'",
                           SCHEMATABLESCORE, col_name, cmp_operator, input_value);
    
    SPI_connect();
    ret = SPI_exec(query, 0);
    
    if (ret != SPI_OK_SELECT) {
        elog(ERROR, "Query operation failed: %s", query);
    }

    // Process results
    for (int i = 0; i < SPI_processed; i++) {
        Datum values[PG_STAT_GET_SCORE_COLS];
        bool nulls[PG_STAT_GET_SCORE_COLS];

        for (int j = 0; j < PG_STAT_GET_SCORE_COLS; j++) {
            values[j] = SPI_getbinval(SPI_tuptable->vals[i], SPI_tuptable->tupdesc, j + 1, &nulls[j]);
        }

        tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, values, nulls);
    }

    // Clean up
    SPI_finish();
    pfree(query);

    return (Datum) 0;
}


